#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include "common.h"
#include "schema.h"
#include "cpulib.h"
#define BLOCK  (1024*1024*50)

extern void tableScan(struct scanNode *,struct statistic *);
extern struct tableNode* hashJoin(struct joinNode *, struct statistic *);
extern struct tableNode* groupBy(struct groupByNode *,struct statistic *);
extern struct tableNode* orderBy(struct orderByNode *, struct statistic *);
extern void materializeCol(struct materializeNode * mn, struct statistic *);
int main(int argc, char ** argv){

//initialize the gpu device
	int * tmp;
	cudaMalloc((void **)&tmp, 4);
	cudaFree(tmp);

	struct statistic pp;
	pp.total = pp.kernel = 0;
	struct tableNode *result = (struct tableNode*) malloc(sizeof(struct tableNode));
	initTable(result);
	struct tableNode *customerTable;
	struct tableNode *supplierTable;
	struct tableNode *ddateTable;
	int outFd;
	int outSize;
	char *outTable;
	long offset;
	struct columnHeader header;

	struct timespec start,end;
	struct timespec tstart,tend;
	double total = 0, memTotal = 0;

	clock_gettime(CLOCK_REALTIME,&start);

	customerTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	customerTable->totalAttr = 2;
	customerTable->attrType = (int *) malloc(sizeof(int)*2);
	customerTable->attrSize = (int *) malloc(sizeof(int)*2);
	customerTable->attrTotalSize = (int *) malloc(sizeof(int)*2);
	customerTable->dataPos = (int *) malloc(sizeof(int)*2);
	customerTable->dataFormat = (int *) malloc(sizeof(int)*2);
	customerTable->content = (char **) malloc(sizeof(char *)*2);
	customerTable->attrSize[0] = 15;
	customerTable->attrType[0] = STRING;
	customerTable->dataPos[0] = MEM;
	outFd = open("CUSTOMER4",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	customerTable->dataFormat[0] = header.format;
	customerTable->tupleNum = header.tupleNum;
	outSize = lseek(outFd,offset,SEEK_END);
	customerTable->attrTotalSize[0] = outSize;
	customerTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(customerTable->content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	customerTable->attrSize[1] = sizeof(int);
	customerTable->attrType[1] = INT;
	customerTable->dataPos[1] = MEM;
	outFd = open("CUSTOMER0",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	customerTable->dataFormat[1] = header.format;
	outSize = lseek(outFd,offset,SEEK_END);
	customerTable->attrTotalSize[1] = outSize;
	customerTable->content[1] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(customerTable->content[1],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	customerTable->tupleSize = 0 + 15 + sizeof(int);

	struct scanNode customerRel;
	customerRel.tn = customerTable;
	customerRel.hasWhere = 1;
	customerRel.whereAttrNum = 1;
	customerRel.whereAttrType = (int *)malloc(sizeof(int)*1);
	customerRel.whereAttrSize = (int *)malloc(sizeof(int)*1);
	customerRel.whereSize = (int *)malloc(sizeof(int)*1);
	customerRel.whereFormat = (int *)malloc(sizeof(int)*1);
	customerRel.content = (char **)malloc(sizeof(char *)*1);
	customerRel.keepInGpu = 1;
	customerRel.whereAttrType[0] = STRING;
	customerRel.whereAttrSize[0] = 12;
	outFd = open("CUSTOMER5",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	customerRel.whereFormat[0] = header.format;
	offset = sizeof(struct columnHeader);
	outSize = lseek(outFd,offset,SEEK_END);
	customerRel.whereSize[0] = outSize;
	customerRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME, &tstart);
	memcpy(customerRel.content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME, &tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize + offset);
	close(outFd);
	customerRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(customerRel.filter)->nested = 0;
	(customerRel.filter)->expNum = 1;
	(customerRel.filter)->exp = (struct whereExp*) malloc(sizeof(struct whereExp) *1);
	(customerRel.filter)->andOr = EXP;
	(customerRel.filter)->exp[0].index = 0;
	(customerRel.filter)->exp[0].relation = EQ;
	strcpy((customerRel.filter)->exp[0].content,"ASIA");

	tableScan(&customerRel, &pp);
	freeScan(&customerRel);

	supplierTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	supplierTable->totalAttr = 2;
	supplierTable->attrType = (int *) malloc(sizeof(int)*2);
	supplierTable->attrSize = (int *) malloc(sizeof(int)*2);
	supplierTable->attrTotalSize = (int *) malloc(sizeof(int)*2);
	supplierTable->dataPos = (int *) malloc(sizeof(int)*2);
	supplierTable->dataFormat = (int *) malloc(sizeof(int)*2);
	supplierTable->content = (char **) malloc(sizeof(char *)*2);
	supplierTable->attrSize[0] = 15;
	supplierTable->attrType[0] = STRING;
	supplierTable->dataPos[0] = MEM;
	outFd = open("SUPPLIER4",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	supplierTable->dataFormat[0] = header.format;
	supplierTable->tupleNum = header.tupleNum;
	outSize = lseek(outFd,offset,SEEK_END);
	supplierTable->attrTotalSize[0] = outSize;
	supplierTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(supplierTable->content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	supplierTable->attrSize[1] = sizeof(int);
	supplierTable->attrType[1] = INT;
	supplierTable->dataPos[1] = MEM;
	outFd = open("SUPPLIER0",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	supplierTable->dataFormat[1] = header.format;
	outSize = lseek(outFd,offset,SEEK_END);
	supplierTable->attrTotalSize[1] = outSize;
	supplierTable->content[1] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(supplierTable->content[1],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	supplierTable->tupleSize = 0 + 15 + sizeof(int);

	struct scanNode supplierRel;
	supplierRel.tn = supplierTable;
	supplierRel.hasWhere = 1;
	supplierRel.whereAttrNum = 1;
	supplierRel.whereAttrType = (int *)malloc(sizeof(int)*1);
	supplierRel.whereAttrSize = (int *)malloc(sizeof(int)*1);
	supplierRel.whereSize = (int *)malloc(sizeof(int)*1);
	supplierRel.whereFormat = (int *)malloc(sizeof(int)*1);
	supplierRel.content = (char **)malloc(sizeof(char *)*1);
	supplierRel.keepInGpu = 1;
	supplierRel.whereAttrType[0] = STRING;
	supplierRel.whereAttrSize[0] = 12;
	outFd = open("SUPPLIER5",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	supplierRel.whereFormat[0] = header.format;
	offset = sizeof(struct columnHeader);
	outSize = lseek(outFd,offset,SEEK_END);
	supplierRel.whereSize[0] = outSize;
	supplierRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME, &tstart);
	memcpy(supplierRel.content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME, &tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize + offset);
	close(outFd);
	supplierRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(supplierRel.filter)->nested = 0;
	(supplierRel.filter)->expNum = 1;
	(supplierRel.filter)->exp = (struct whereExp*) malloc(sizeof(struct whereExp) *1);
	(supplierRel.filter)->andOr = EXP;
	(supplierRel.filter)->exp[0].index = 0;
	(supplierRel.filter)->exp[0].relation = EQ;
	strcpy((supplierRel.filter)->exp[0].content,"ASIA");

	tableScan(&supplierRel, &pp);
	freeScan(&supplierRel);

	ddateTable = (struct tableNode *) malloc(sizeof(struct tableNode));
	ddateTable->totalAttr = 2;
	ddateTable->attrType = (int *) malloc(sizeof(int)*2);
	ddateTable->attrSize = (int *) malloc(sizeof(int)*2);
	ddateTable->attrTotalSize = (int *) malloc(sizeof(int)*2);
	ddateTable->dataPos = (int *) malloc(sizeof(int)*2);
	ddateTable->dataFormat = (int *) malloc(sizeof(int)*2);
	ddateTable->content = (char **) malloc(sizeof(char *)*2);
	ddateTable->attrSize[0] = sizeof(int);
	ddateTable->attrType[0] = INT;
	ddateTable->dataPos[0] = MEM;
	outFd = open("DDATE4",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	ddateTable->dataFormat[0] = header.format;
	ddateTable->tupleNum = header.tupleNum;
	outSize = lseek(outFd,offset,SEEK_END);
	ddateTable->attrTotalSize[0] = outSize;
	ddateTable->content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(ddateTable->content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	ddateTable->attrSize[1] = sizeof(int);
	ddateTable->attrType[1] = INT;
	ddateTable->dataPos[1] = MEM;
	outFd = open("DDATE0",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	offset = sizeof(struct columnHeader);
	ddateTable->dataFormat[1] = header.format;
	outSize = lseek(outFd,offset,SEEK_END);
	ddateTable->attrTotalSize[1] = outSize;
	ddateTable->content[1] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset ,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME,&tstart);
	memcpy(ddateTable->content[1],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME,&tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize+offset);
	close(outFd);
	ddateTable->tupleSize = 0 + sizeof(int) + sizeof(int);

	struct scanNode ddateRel;
	ddateRel.tn = ddateTable;
	ddateRel.hasWhere = 1;
	ddateRel.whereAttrNum = 1;
	ddateRel.whereAttrType = (int *)malloc(sizeof(int)*2);
	ddateRel.whereAttrSize = (int *)malloc(sizeof(int)*2);
	ddateRel.whereSize = (int *)malloc(sizeof(int)*2);
	ddateRel.whereFormat = (int *)malloc(sizeof(int)*2);
	ddateRel.content = (char **)malloc(sizeof(char *)*2);
	ddateRel.keepInGpu = 1;
	ddateRel.whereAttrType[0] = INT;
	ddateRel.whereAttrSize[0] = sizeof(int);
	outFd = open("DDATE4",O_RDONLY);
	read(outFd,&header, sizeof(struct columnHeader));
	ddateRel.whereFormat[0] = header.format;
	offset = sizeof(struct columnHeader);
	outSize = lseek(outFd,offset,SEEK_END);
	ddateRel.whereSize[0] = outSize;
	ddateRel.content[0] = (char *) malloc(outSize);
	outTable =(char *) mmap(0,outSize + offset,PROT_READ,MAP_SHARED,outFd,0);

	clock_gettime(CLOCK_REALTIME, &tstart);
	memcpy(ddateRel.content[0],outTable + offset,outSize);
	clock_gettime(CLOCK_REALTIME, &tend);
	memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

	munmap(outTable,outSize + offset);
	close(outFd);
	ddateRel.filter = (struct whereCondition *)malloc(sizeof(struct whereCondition));
	(ddateRel.filter)->nested = 0;
	(ddateRel.filter)->expNum = 2;
	(ddateRel.filter)->exp = (struct whereExp*) malloc(sizeof(struct whereExp) *2);
	(ddateRel.filter)->andOr = AND;
	(ddateRel.filter)->exp[0].index = 0;
	(ddateRel.filter)->exp[0].relation = GEQ;
	{
		int tmp = 1992;
		memcpy((ddateRel.filter)->exp[0].content, &tmp,sizeof(int));
	}
	(ddateRel.filter)->exp[1].index = 0;
	(ddateRel.filter)->exp[1].relation = LEQ;
	{
		int tmp = 1997;
		memcpy((ddateRel.filter)->exp[1].content, &tmp,sizeof(int));
	}
	tableScan(&ddateRel, &pp);
	freeScan(&ddateRel);

	struct tableNode *lineorderTable = (struct tableNode*)malloc(sizeof(struct tableNode));
	lineorderTable->totalAttr = 4;
	lineorderTable->attrType = (int *) malloc(sizeof(int)*4);
	lineorderTable->attrSize = (int *) malloc(sizeof(int)*4);
	lineorderTable->attrTotalSize = (int *) malloc(sizeof(int)*4);
	lineorderTable->dataPos = (int *) malloc(sizeof(int)*4);
	lineorderTable->dataFormat = (int *) malloc(sizeof(int)*4);
	lineorderTable->content = (char **) malloc(sizeof(char *)*4);
	outFd = open("LINEORDER12",O_RDONLY);
	read(outFd,&header,sizeof(columnHeader));
	lineorderTable->tupleNum = header.tupleNum;
	close(outFd);
	lineorderTable->attrType[0] = FLOAT;
	lineorderTable->attrSize[0] = sizeof(float);
	lineorderTable->dataPos[0] = MEM;
	lineorderTable->attrType[1] = INT;
	lineorderTable->attrSize[1] = sizeof(int);
	lineorderTable->dataPos[1] = MEM;
	lineorderTable->attrType[2] = INT;
	lineorderTable->attrSize[2] = sizeof(int);
	lineorderTable->dataPos[2] = MEM;
	lineorderTable->attrType[3] = INT;
	lineorderTable->attrSize[3] = sizeof(int);
	lineorderTable->dataPos[3] = MEM;
	lineorderTable->tupleSize = 0 + sizeof(float) + sizeof(int) + sizeof(int) + sizeof(int);

	struct joinNode jNode;
	jNode.dimNum = 3;
	jNode.factTable = lineorderTable;
	jNode.dimTable = (struct tableNode **) malloc(sizeof(struct tableNode) * jNode.dimNum);
	jNode.factIndex = (int *) malloc(sizeof(int) * jNode.dimNum);
	jNode.dimIndex = (int *) malloc(sizeof(int) * jNode.dimNum);

	jNode.dimIndex[0] = 1;
	jNode.factIndex[0] = 3;
	jNode.dimTable[0] = customerTable;

	jNode.dimIndex[1] = 1;
	jNode.factIndex[1] = 2;
	jNode.dimTable[1] = supplierTable;

	jNode.dimIndex[2] = 1;
	jNode.factIndex[2] = 1;
	jNode.dimTable[2] = ddateTable;

	jNode.totalAttr = 4;
	jNode.keepInGpu = (int *) malloc(sizeof(int) * 4);
	for(int k=0;k<4;k++)
		jNode.keepInGpu[k] = 1;

	jNode.attrType = (int *) (malloc(sizeof(int) * jNode.totalAttr));
	jNode.attrSize = (int *) (malloc(sizeof(int) * jNode.totalAttr));
	jNode.attrType[0] = STRING;
	jNode.attrSize[0] = 15;
	jNode.attrType[1] = STRING;
	jNode.attrSize[1] = 15;
	jNode.attrType[2] = INT;
	jNode.attrSize[2] = sizeof(int);
	jNode.attrType[3] = FLOAT;
	jNode.attrSize[3] = sizeof(float);
	jNode.tupleSize = 0+15+15+sizeof(int)+sizeof(float);

	jNode.factOutputNum = 1;
	jNode.factOutputIndex = (int *) malloc(jNode.factOutputNum * sizeof(int));
	jNode.factOutputPos = (int *) malloc(jNode.factOutputNum * sizeof(int));
	jNode.factOutputIndex[0] = 0;
	jNode.factOutputPos[0] = 3;
	jNode.dimOutputTotal = 3;
	jNode.dimOutputNum = (int *) malloc( sizeof(int) * jNode.dimNum);
	jNode.dimOutputIndex = (int **) malloc( sizeof(int*) * jNode.dimNum);
	jNode.dimOutputPos = (int *) malloc( sizeof(int) * jNode.dimOutputTotal);
	jNode.dimOutputNum[0] = 1;
	jNode.dimOutputIndex[0] = (int *) malloc(sizeof(int) *1);
	jNode.dimOutputIndex[0][0] = 0;
	jNode.dimOutputNum[1] = 1;
	jNode.dimOutputIndex[1] = (int *) malloc(sizeof(int) *1);
	jNode.dimOutputIndex[1][0] = 0;
	jNode.dimOutputNum[2] = 1;
	jNode.dimOutputIndex[2] = (int *) malloc(sizeof(int) *1);
	jNode.dimOutputIndex[2][0] = 0;
	jNode.dimOutputPos[0] = 0;
	jNode.dimOutputPos[1] = 1;
	jNode.dimOutputPos[2] = 2;
	int pass = lineorderTable->tupleNum / BLOCK + 1;
	long tupleUnit = lineorderTable->tupleNum / pass;
	long nextScan = tupleUnit;
	long restTuple = lineorderTable->tupleNum;
	long tupleOffset = 0;
	offset = 0;

	for(int i=0;i<pass;i++){

		if(restTuple < nextScan)
			nextScan = restTuple;

		lineorderTable->tupleNum = nextScan;
		outFd = open("LINEORDER12", O_RDONLY);
		read(outFd, &header, sizeof(struct columnHeader));
		lineorderTable->dataFormat[0] = header.format;
		if(header.format == UNCOMPRESSED){
			offset = tupleOffset *sizeof(float)+sizeof(struct columnHeader);
			outSize = nextScan*sizeof(float);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[0] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[0], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[0] = outSize;
		}else if (header.format == DICT){
			struct dictHeader dheader;
			read(outFd, &dheader, sizeof(struct dictHeader));
			offset = tupleOffset * dheader.bitNum / 8 + sizeof(struct columnHeader) + sizeof(struct dictHeader);
			outSize = nextScan * dheader.bitNum / 8 + sizeof(struct dictHeader);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[0] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[0], &dheader, sizeof(struct dictHeader));
			memcpy(lineorderTable->content[0] + sizeof(struct dictHeader), outTable+offset, outSize - sizeof(struct dictHeader));
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[0] = outSize;
		}else if (header.format == RLE){
			offset = sizeof(struct columnHeader);
			outSize = lseek(outFd, 0, SEEK_END) - offset;
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[0] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[0], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[0] = outSize;
		}
		outFd = open("LINEORDER5", O_RDONLY);
		read(outFd, &header, sizeof(struct columnHeader));
		lineorderTable->dataFormat[1] = header.format;
		if(header.format == UNCOMPRESSED){
			offset = tupleOffset *sizeof(int)+sizeof(struct columnHeader);
			outSize = nextScan*sizeof(int);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[1] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[1], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[1] = outSize;
		}else if (header.format == DICT){
			struct dictHeader dheader;
			read(outFd, &dheader, sizeof(struct dictHeader));
			offset = tupleOffset * dheader.bitNum / 8 + sizeof(struct columnHeader) + sizeof(struct dictHeader);
			outSize = nextScan * dheader.bitNum / 8 + sizeof(struct dictHeader);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[1] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[1], &dheader, sizeof(struct dictHeader));
			memcpy(lineorderTable->content[1] + sizeof(struct dictHeader), outTable+offset, outSize - sizeof(struct dictHeader));
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[1] = outSize;
		}else if (header.format == RLE){
			offset = sizeof(struct columnHeader);
			outSize = lseek(outFd, 0, SEEK_END) - offset;
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[1] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[1], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[1] = outSize;
		}
		outFd = open("LINEORDER4", O_RDONLY);
		read(outFd, &header, sizeof(struct columnHeader));
		lineorderTable->dataFormat[2] = header.format;
		if(header.format == UNCOMPRESSED){
			offset = tupleOffset *sizeof(int)+sizeof(struct columnHeader);
			outSize = nextScan*sizeof(int);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[2] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[2], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[2] = outSize;
		}else if (header.format == DICT){
			struct dictHeader dheader;
			read(outFd, &dheader, sizeof(struct dictHeader));
			offset = tupleOffset * dheader.bitNum / 8 + sizeof(struct columnHeader) + sizeof(struct dictHeader);
			outSize = nextScan * dheader.bitNum / 8 + sizeof(struct dictHeader);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[2] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[2], &dheader, sizeof(struct dictHeader));
			memcpy(lineorderTable->content[2] + sizeof(struct dictHeader), outTable+offset, outSize - sizeof(struct dictHeader));
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[2] = outSize;
		}else if (header.format == RLE){
			offset = sizeof(struct columnHeader);
			outSize = lseek(outFd, 0, SEEK_END) - offset;
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[2] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[2], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[2] = outSize;
		}
		outFd = open("LINEORDER2", O_RDONLY);
		read(outFd, &header, sizeof(struct columnHeader));
		lineorderTable->dataFormat[3] = header.format;
		if(header.format == UNCOMPRESSED){
			offset = tupleOffset *sizeof(int)+sizeof(struct columnHeader);
			outSize = nextScan*sizeof(int);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[3] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[3], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[3] = outSize;
		}else if (header.format == DICT){
			struct dictHeader dheader;
			read(outFd, &dheader, sizeof(struct dictHeader));
			offset = tupleOffset * dheader.bitNum / 8 + sizeof(struct columnHeader) + sizeof(struct dictHeader);
			outSize = nextScan * dheader.bitNum / 8 + sizeof(struct dictHeader);
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[3] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[3], &dheader, sizeof(struct dictHeader));
			memcpy(lineorderTable->content[3] + sizeof(struct dictHeader), outTable+offset, outSize - sizeof(struct dictHeader));
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[3] = outSize;
		}else if (header.format == RLE){
			offset = sizeof(struct columnHeader);
			outSize = lseek(outFd, 0, SEEK_END) - offset;
			outTable = (char *)mmap(0,outSize+offset,PROT_READ,MAP_SHARED,outFd,0);
			lineorderTable->content[3] = (char *) malloc(outSize);

			clock_gettime(CLOCK_REALTIME, &tstart);
			memcpy(lineorderTable->content[3], outTable+offset, outSize);
			clock_gettime(CLOCK_REALTIME, &tend);
			memTotal += (tend.tv_sec -  tstart.tv_sec)* BILLION + tend.tv_nsec - tstart.tv_nsec;

			munmap(outTable,outSize + offset);
			close(outFd);
			lineorderTable->attrTotalSize[3] = outSize;
		}
		struct tableNode *join1 = hashJoin(&jNode, &pp);
		if(lineorderTable->dataPos[0] == MEM)
			free(lineorderTable->content[0]);
		else
			cudaFree(lineorderTable->content[0]);
		if(lineorderTable->dataPos[1] == MEM)
			free(lineorderTable->content[1]);
		else
			cudaFree(lineorderTable->content[1]);
		if(lineorderTable->dataPos[2] == MEM)
			free(lineorderTable->content[2]);
		else
			cudaFree(lineorderTable->content[2]);
		if(lineorderTable->dataPos[3] == MEM)
			free(lineorderTable->content[3]);
		else
			cudaFree(lineorderTable->content[3]);
		mergeIntoTable(result,join1,&pp);
		freeTable(join1);
		tupleOffset +=tupleUnit;
		restTuple -= nextScan;
	}

	struct groupByNode * gbNode = (struct groupByNode *) malloc(sizeof(struct groupByNode));
	gbNode->table = result;
	gbNode->groupByColNum = 3;
	gbNode->groupByIndex = (int *)malloc(sizeof(int) * 3);
	gbNode->groupByType = (int *)malloc(sizeof(int) * 3);
	gbNode->groupBySize = (int *)malloc(sizeof(int) * 3);
	gbNode->groupByIndex[0] = 0;
	gbNode->groupByType[0] = gbNode->table->attrType[0];
	gbNode->groupBySize[0] = gbNode->table->attrSize[0];
	gbNode->groupByIndex[1] = 1;
	gbNode->groupByType[1] = gbNode->table->attrType[1];
	gbNode->groupBySize[1] = gbNode->table->attrSize[1];
	gbNode->groupByIndex[2] = 2;
	gbNode->groupByType[2] = gbNode->table->attrType[2];
	gbNode->groupBySize[2] = gbNode->table->attrSize[2];
	gbNode->outputAttrNum = 4;
	gbNode->attrType = (int *) malloc(sizeof(int) *4);
	gbNode->attrSize = (int *) malloc(sizeof(int) *4);
	gbNode->tupleSize = 0;
	gbNode->gbExp = (struct groupByExp *) malloc(sizeof(struct groupByExp) * 4);
	gbNode->attrType[0] = result->attrType[0];
	gbNode->attrSize[0] = result->attrSize[0];
	gbNode->tupleSize += result->attrSize[0];
	gbNode->gbExp[0].func = NOOP;
	gbNode->gbExp[0].exp.op = NOOP;
	gbNode->gbExp[0].exp.exp = NULL;
	gbNode->gbExp[0].exp.opNum = 1;
	gbNode->gbExp[0].exp.opType = COLUMN;
	gbNode->gbExp[0].exp.opValue = 0;
	gbNode->attrType[1] = result->attrType[1];
	gbNode->attrSize[1] = result->attrSize[1];
	gbNode->tupleSize += result->attrSize[1];
	gbNode->gbExp[1].func = NOOP;
	gbNode->gbExp[1].exp.op = NOOP;
	gbNode->gbExp[1].exp.exp = NULL;
	gbNode->gbExp[1].exp.opNum = 1;
	gbNode->gbExp[1].exp.opType = COLUMN;
	gbNode->gbExp[1].exp.opValue = 1;
	gbNode->attrType[2] = result->attrType[2];
	gbNode->attrSize[2] = result->attrSize[2];
	gbNode->tupleSize += result->attrSize[2];
	gbNode->gbExp[2].func = NOOP;
	gbNode->gbExp[2].exp.op = NOOP;
	gbNode->gbExp[2].exp.exp = NULL;
	gbNode->gbExp[2].exp.opNum = 1;
	gbNode->gbExp[2].exp.opType = COLUMN;
	gbNode->gbExp[2].exp.opValue = 2;
	gbNode->tupleSize += sizeof(float);
	gbNode->attrType[3] = FLOAT;
	gbNode->attrSize[3] = sizeof(float);
	gbNode->gbExp[3].func = SUM;
	gbNode->gbExp[3].exp.op = NOOP;
	gbNode->gbExp[3].exp.opNum = 1;
	gbNode->gbExp[3].exp.exp = NULL;
	gbNode->gbExp[3].exp.opType = COLUMN;
	gbNode->gbExp[3].exp.opValue = 3;
	struct tableNode * gbResult = groupBy(gbNode, &pp);
	freeGroupByNode(gbNode);

	struct orderByNode * odNode = (struct orderByNode *) malloc(sizeof(struct orderByNode));
	odNode->table = gbResult;
	odNode->orderByNum = 2;
	odNode->orderBySeq = (int *) malloc(sizeof(int) * odNode->orderByNum);
	odNode->orderByIndex = (int *) malloc(sizeof(int) * odNode->orderByNum);
	odNode->orderBySeq[0] = ASC;
	odNode->orderByIndex[0] = 2;
	odNode->orderBySeq[1] = DESC;
	odNode->orderByIndex[1] = 3;
	struct tableNode * odResult = orderBy(odNode,&pp);
	freeOrderByNode(odNode);

	struct materializeNode mn;
	mn.table = odResult;
	materializeCol(&mn, &pp);
	freeTable(odResult);

	clock_gettime(CLOCK_REALTIME,&end);

	total = (end.tv_sec -  start.tv_sec)* BILLION + end.tv_nsec - start.tv_nsec;
	pp.total = total - memTotal;
	printf("kernel:%lf total:%lf\n",pp.kernel,pp.total);
}

